/**************************************************************************
 *
 * Copyright (C) 2007 University of California, Los Angeles.
 *
 * This program is free software; you can redistribute it and/or modify
 * it under the terms of the GNU General Public License as published by
 * the Free Software Foundation; either version 2 of the License, or
 * (at your option) any later version.
 *
 * This program is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 * GNU General Public License for more details.
 *
 * You should have received a copy of the GNU General Public License
 * along with this program; if not, write to the Free Software
 * Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
 *
 *************************************************************************/

/**
 * 
 */
package edu.ucla.cs.typecast.test;

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.io.OutputStream;
import java.io.PrintStream;
import java.net.InetAddress;
import java.util.Collections;
import java.util.Date;
import java.util.HashMap;
import java.util.Map;
import java.util.Timer;
import java.util.TimerTask;

import edu.ucla.cs.typecast.discovery.ServiceAdvertisement;
import edu.ucla.cs.typecast.discovery.ServiceDiscoveryClient;
import edu.ucla.cs.typecast.discovery.ServiceDiscoveryListener;
import edu.ucla.cs.typecast.discovery.ServiceDiscoveryManager;
import edu.ucla.cs.typecast.discovery.ServiceTemplate;
import edu.ucla.cs.typecast.net.EndPoint;
import edu.ucla.cs.typecast.net.NetUtil;
import edu.ucla.cs.typecast.rmi.TypeCastClient;
import edu.ucla.cs.typecast.rmi.TypeCastServer;

/**
 *
 * @date Nov 12, 2007
 */
public class DiscoverSender implements ServiceDiscoveryListener {
	private static final Object logLock = new Object(); 
	
	private long delay; 
	private String sessionID; 
	private long duration; 
	private int rate; 
	private long interval; 
	private EndPoint serviceEndPoint; 
	private int payloadSize; 
	private OutputStream ostream; 
	
	private Ping pingClient; 
	private TypeCastServer ackServer; 

	private long seqno = 1; 
	private byte[] payload; 
	private long totalSend = 0; 
	private long totalAcked = 0; 
	private double totalLatency = 0; 
	private double aveLatency = 0; 	
	private Timer timer = new Timer(); 
	private boolean init = false; 
	private Class targetType; 
	private String targetDisplayType; 
	private long startEffectiveCount = -1; 
	private long totalEffectiveCount = 0; 
	private double totalEffectiveLatency = 0; 
	private double aveEffectiveLatency = 0; 
	private PrintStream latencyOutStream; 
	
	private Map<Long, Date> sendTimeMap = Collections.synchronizedMap(new HashMap<Long, Date>(1024));
	TimerTask sendTask; 
	
	public DiscoverSender(Class targetType, EndPoint endPoint, String sessionID, long delay, long duration, int rate, int payloadSize, OutputStream output, OutputStream latencyStream) { 
		this.targetType = targetType; 
		this.targetDisplayType = getDisplayName(targetType); 
		this.delay = delay; 
		this.serviceEndPoint = endPoint; 
		this.sessionID = sessionID; 
		this.duration = duration; 
		this.rate = rate; 
		this.payloadSize = payloadSize; 
		this.interval = 1000 / rate; 		
		this.payload = new byte[payloadSize]; 
		this.ostream = output; 
		this.startEffectiveCount = rate * 60; // count after 1 minute; 
		if (latencyStream != null) { 
			this.latencyOutStream = new PrintStream(latencyStream); 
		}
		start(); 
	}
	
	private void start() { 
		try { 
			TimerTask cancelTask = new TimerTask() { 
				public void run() { 
					try { 
						sendTask.cancel(); 
					} catch(Throwable t) { 
						t.printStackTrace(); 
					}
				}
			}; 
			timer.schedule(cancelTask, delay + duration); 			
			
			
			// Schedule timer to send packets
			sendTask = new TimerTask() { 
				public void run() { 
					try { 
						send(); 
					} catch(Throwable t) { 
						t.printStackTrace(); 
					}
				}
			}; 
			
			timer.scheduleAtFixedRate(sendTask, delay, interval); 
			
			TimerTask stopTask = new TimerTask() { 
				public void run() { 
					try { 
						timer.cancel(); 
						shutdown(); 
					} catch(Throwable t) { 
						t.printStackTrace(); 
					}
					try { 
						PrintStream logStream = new PrintStream(ostream); 
						saveStatistics(logStream); 
						if(latencyOutStream != null) { 
							latencyOutStream.flush(); 
						}						
					} catch(Throwable t) { 
						t.printStackTrace(); 
					}
				}
			};
			// Stop everthing 5 minutes after the sending task is done. 
			timer.schedule(stopTask, delay + duration + 1 * 60 * 1000); 
			
		} catch(Throwable t) { 
			t.printStackTrace(); 
		}
	}
	
	/* (non-Javadoc)
	 * @see edu.ucla.cs.typecast.test.PingAck#pong(java.net.InetAddress, java.lang.String, java.lang.String)
	 */
	public void notify(ServiceAdvertisement serviceInfo) {  
		// TODO Auto-generated method stub
		Date sendTime = this.sendTimeMap.get(new Long(serviceInfo.getSeqno())); 
		if (sendTime == null) { 
			System.out.println("Error: missing send timestamp for session " + sessionID + " seqno=" + serviceInfo.getSeqno()); 
			return; 
		}
		totalAcked++; 
		Date receiveTime = new Date(); 
		this.totalLatency += receiveTime.getTime() - sendTime.getTime(); 
		this.aveLatency = totalLatency / totalAcked; 	
		if (serviceInfo.getSeqno() > this.startEffectiveCount) { 
			this.totalEffectiveCount++; 
			this.totalEffectiveLatency += receiveTime.getTime() - sendTime.getTime();
			this.aveEffectiveLatency = this.totalEffectiveLatency / this.totalEffectiveCount; 
		}
		
		if(this.latencyOutStream != null) { 
			latencyOutStream.print(targetDisplayType); 
			latencyOutStream.print(" "); 
			latencyOutStream.print(this.serviceEndPoint); 
			latencyOutStream.print(" "); 
			latencyOutStream.print(NetUtil.getLocalHost().getHostAddress()); 
			latencyOutStream.print(" "); 
			latencyOutStream.print(this.sessionID); 
			latencyOutStream.print(" "); 
			latencyOutStream.print(serviceInfo.getSeqno()); 
			latencyOutStream.print(" "); 				
			latencyOutStream.print(receiveTime.getTime() - sendTime.getTime()); 
			latencyOutStream.print(""); 
			latencyOutStream.println(); 
		}		
		System.out.println("Receive TypeAdvertisement from " + serviceInfo.getHost().getHostAddress() + " seqno = " + serviceInfo.getSeqno() + " at " + new Date());  
	}
	
	private synchronized void send() { 
		if (!init) { 
			init(); 
		}
		try { 
			sendTimeMap.put(new Long(seqno), new Date()); 
	        Class[] types = new Class[] { targetType }; 
	        ServiceTemplate template = new ServiceTemplate(types, seqno); 
			seqno++; 	        
	        ServiceDiscoveryClient client = new ServiceDiscoveryClient(this.serviceEndPoint); 
	        client.discover(types, this.serviceEndPoint, template); 
	        totalSend++; 
			System.out.println("Sending discovery request to " + this.targetDisplayType + " with seqno " + (seqno-1) + " at " + new Date());  	        
		} catch(Throwable t) { 
			t.printStackTrace(); 
		}
	        
	}
	
	private void init() { 
		try { 
            Class[] types = new Class[] { targetType }; 
            ServiceTemplate template = new ServiceTemplate(types); 
            ServiceDiscoveryManager.getInstance().addDiscoveryListener(this.serviceEndPoint, template, this); 
			init = true; 
		} catch(Throwable t) { 
			t.printStackTrace(); 
		}
	}
	
	private void shutdown() { 
        Class[] types = new Class[] { targetType }; 
        ServiceTemplate template = new ServiceTemplate(types); 
        ServiceDiscoveryManager.getInstance().removeDiscoveryListener(this); 
	}

	protected TypeCastServer createTypeCastServerForAck() throws IOException { 
		return  new TypeCastServer(null); 		
	}
	
	public void saveStatistics(PrintStream out) { 		
		try { 
			synchronized(logLock) { 
				Date now = new Date(); 				
				out.print(targetDisplayType); 
				out.print(" "); 
				out.print(this.serviceEndPoint); 
				out.print(" "); 
				out.print(NetUtil.getLocalHost().getHostAddress()); 
				out.print(" "); 
				out.print(this.sessionID); 
				out.print(" "); 
				out.print(this.duration/1000); 
				out.print(" "); 
				out.print(this.rate); 
				out.print(" "); 
				out.print(this.totalSend); 
				out.print(" "); 
				out.print(this.totalAcked); 
				out.print(" "); 
				out.print(((double)totalAcked) / totalSend); 
				out.print(" "); 
				out.print(this.aveLatency);
				out.print(" "); 
				out.print(this.aveEffectiveLatency); 
				out.print(" "); 
				out.print(now); 
				out.println(""); 
				out.flush(); 
			} 
		} catch(Throwable t) { 
			t.printStackTrace(); 
		}
	}
	
	public String getDisplayName(Class type) { 
		String className = type.getName(); 
		return className.substring("edu.ucla.cs.typecast.test.Ping".length()); 
	}
}
